<?php

namespace app\process;

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
use app\model\Room;
use app\model\User;
use app\model\RoomUser;
use app\service\RedisService;

class WebSocketServer
{
    private $rooms = []; // 房间连接映射（本进程内的连接对象）
    private $users = []; // 用户连接映射（本进程内的连接对象）
    private $redis;
    private $userRooms = []; // 用户房间映射（本进程内）
    private $connectionUsers = []; // 连接到用户ID的映射（本进程内）
    private $workerId; // 当前 worker 进程ID

    public function __construct()
    {
        $this->redis = RedisService::getInstance();
        global $argv;
        $this->workerId = isset($argv[2]) ? $argv[2] : 0;
    }
    
    /**
     * Redis Key 管理
     */
    private function getRoomUsersKey($roomId) { return "ws:room:{$roomId}:users"; }
    private function getUserWorkerKey($userId) { return "ws:user:{$userId}:worker"; }
    private function getUserRoomsKey($userId) { return "ws:user:{$userId}:rooms"; }

    public function start()
    {
        $worker = new Worker('websocket://0.0.0.0:8282');
        $worker->count = 4;
        
        $worker->onConnect = [$this, 'onConnect'];
        $worker->onMessage = [$this, 'onMessage'];
        $worker->onClose = [$this, 'onClose'];
        $worker->onError = [$this, 'onError'];
        
        // 添加定时器，定期检查Redis中的消息
        $worker->onWorkerStart = [$this, 'onWorkerStart'];
        
        Worker::runAll();
    }

    /**
     * Worker启动时
     * @param Worker $worker Worker实例
     */
    public function onWorkerStart($worker)
    {
        global $argv;
        $workerId = isset($argv[2]) ? $argv[2] : 0;
        $timestamp = date('Y-m-d H:i:s');
        $processId = getmypid() ?? '未知';
        
        // 更新 workerId（从全局参数获取）
        $this->workerId = $workerId;
        
        $logMessage = "\n" .
            "========================================\n" .
            "========== WebSocket Worker {$workerId} 启动 ==========\n" .
            "========================================\n" .
            "时间戳: {$timestamp}\n" .
            "进程ID: {$processId}\n" .
            "Worker ID: {$workerId}\n" .
            "监听地址: " . ($worker->listen ?? 'websocket://0.0.0.0:8282') . "\n" .
            "Worker 名称: " . ($worker->name ?? 'websocket') . "\n" .
            "Worker 进程数: " . ($worker->count ?? 1) . "\n" .
            "Worker {$workerId} 启动 Redis 消息轮询（每3秒检查一次）\n" .
            "========================================\n" .
            "等待客户端连接...\n" .
            "连接事件回调已绑定:\n" .
            "  - onConnect: " . (method_exists($this, 'onConnect') ? '✅' : '❌') . "\n" .
            "  - onMessage: " . (method_exists($this, 'onMessage') ? '✅' : '❌') . "\n" .
            "  - onClose: " . (method_exists($this, 'onClose') ? '✅' : '❌') . "\n" .
            "  - onError: " . (method_exists($this, 'onError') ? '✅' : '❌') . "\n" .
            "========================================\n";
        
        // 输出到控制台
        echo $logMessage;
        
        // 同时记录到日志文件
        \Workerman\Worker::log($logMessage);
        
        // 关键修改：所有 worker 进程都使用轮询机制接收 Redis 消息
        // 因为每个 worker 进程有独立的连接，都需要接收消息来广播给本进程内的连接
        // 如果只有 worker 0 订阅，其他 worker 的连接无法收到消息
        // 使用极短的轮询间隔（0.1秒=100ms）以应对高吞吐量场景（每秒1000条消息）
        \Workerman\Timer::add(0.1, [$this, 'checkRedisMessagesFallback']);
        
        // 注意：如果使用 Redis PUB/SUB，只有 worker 0 能接收（因为 PUB/SUB 是广播模式）
        // 但我们的消息需要通过 room_id 分发到对应的 worker，所以使用轮询更合适
        // 如果需要更高性能，可以考虑使用 Redis Streams 或 Channel
        
        // 强制刷新输出缓冲区
        if (function_exists('flush')) {
            flush();
        }
    }

    /**
     * 启动Redis订阅（在单独的线程中）
     */
    private function startRedisSubscription()
    {
        try {
            echo "启动Redis发布订阅机制...\n";
            
            // 创建一个新的Redis连接用于订阅
            $subRedis = RedisService::getNewInstance()->getClient();
            
            // 由于Windows环境限制，我们将直接使用定时器处理Redis订阅
            // 在Windows中使用轮询方式代替阻塞式订阅
            // 注意：使用非阻塞的 rpop 而不是 brPop，避免阻塞事件循环
            \Workerman\Timer::add(1, function() use ($subRedis) {
                try {
                    // 使用非阻塞的 rpop，每次只处理少量消息
                    $maxMessages = 10;
                    for ($i = 0; $i < $maxMessages; $i++) {
                        $message = $subRedis->rpop('room_messages_queue');
                        if (!$message) {
                            break; // 队列为空，退出
                        }
                        $this->handleRedisMessage($message);
                    }
                } catch (\Exception $e) {
                    echo "Redis轮询异常: " . $e->getMessage() . "\n";
                    // 轮询失败时，尝试重新开始轮询
                    // 创建一个新的定时器，避免闭包内的$this作用域问题
                    $server = $this;
                    \Workerman\Timer::add(5, function() use ($server) {
                        $server->startRedisSubscription();
                    }, [], false);
                }
            });
            
            echo "Redis发布订阅机制已启动\n";
        } catch (\Exception $e) {
            echo "Redis订阅启动失败: " . $e->getMessage() . "\n";
            echo "异常文件: " . $e->getFile() . "\n";
            echo "异常行号: " . $e->getLine() . "\n";
            echo "堆栈跟踪: " . $e->getTraceAsString() . "\n";
            
            // 订阅失败时，启动轮询作为备用
            echo "启动轮询作为备用机制\n";
            \Workerman\Timer::add(1, [$this, 'checkRedisMessagesFallback']);
        }
    }

    /**
     * 处理Redis消息（高性能版本，减少日志输出）
     */
    private function handleRedisMessage($message)
    {
        try {
            // 高性能优化：减少日志输出，只在错误时记录
            $data = json_decode($message, true);
            if (!$data || !isset($data['room_id'])) {
                \Workerman\Worker::log("无效的Redis消息格式: " . substr($message, 0, 200));
                return;
            }

            // 确保房间ID是字符串类型，以保证类型一致性
            $roomId = (string)$data['room_id'];
            $messageData = $data['message'];
            
            // 验证消息数据
            if (!is_array($messageData)) {
                \Workerman\Worker::log("消息数据格式错误，房间ID: {$roomId}");
                return;
            }
            
            // 确保房间ID是字符串类型（与 handleUserConnect 中保持一致）
            $roomId = (string)$roomId;
            
            // 快速检查房间是否存在（优化：减少不必要的操作）
            if (!isset($this->rooms[$roomId])) {
                // 尝试数字格式
                $numericRoomId = (string)intval($roomId);
                if ($numericRoomId !== $roomId && isset($this->rooms[$numericRoomId])) {
                    $roomId = $numericRoomId;
                } else {
                    // 仅在非高频情况下尝试自动订阅（避免频繁查询数据库）
                    static $lastAutoSubscribeTime = [];
                    $now = time();
                    if (!isset($lastAutoSubscribeTime[$roomId]) || ($now - $lastAutoSubscribeTime[$roomId]) > 5) {
                        // 5秒内只尝试一次自动订阅，避免频繁查询
                        $lastAutoSubscribeTime[$roomId] = $now;
                        $this->autoSubscribeRoomUsers($roomId);
                    }
                    
                    // 如果仍然不存在，直接返回（可能在别的worker进程）
                    if (!isset($this->rooms[$roomId])) {
                        return; // 静默返回，不记录日志（正常情况）
                    }
                }
            }
            
            // 直接广播消息到当前进程内的连接（高性能版本）
            // 注意：在多进程环境下，每个进程只能广播到自己进程内的连接
            $this->broadcastToRoom($roomId, $messageData);
            
        } catch (\Exception $e) {
            // 仅在异常时记录日志
            \Workerman\Worker::log("处理Redis消息失败: " . $e->getMessage() . ", 文件: " . $e->getFile() . ", 行号: " . $e->getLine());
        }
    }

    /**
     * 备用轮询模式（仅在订阅失败时使用）
     * 使用队列方式，避免阻塞事件循环
     */
    public function checkRedisMessagesFallback()
    {
        try {
            // 优先检查队列消息（非阻塞）
            $this->checkRedisQueue();
            
            // 然后检查房间消息（限制每次处理的数量，避免阻塞）
            $this->checkRoomMessages();
        } catch (\Exception $e) {
            echo "备用轮询模式失败: " . $e->getMessage() . "\n";
            \Workerman\Worker::log("备用轮询模式失败: " . $e->getMessage());
        }
    }
    
    /**
     * 检查 Redis 队列消息（高吞吐量优化版本）
     * 支持每秒处理1000+条消息
     */
    private function checkRedisQueue()
    {
        try {
            // 先检查队列长度，动态调整处理数量
            $queueLength = $this->redis->getClient()->llen('room_messages_queue');
            
            if ($queueLength === 0) {
                return; // 队列为空，直接返回
            }
            
            // 根据队列长度动态调整处理数量（应对高负载场景）：
            // - 队列极长（>1000）：批量处理500条（快速清空）
            // - 队长（>500）：处理300条
            // - 队列中等（>100）：处理100条
            // - 队列短（>10）：处理50条
            // - 队列很短（<10）：处理10条
            if ($queueLength > 1000) {
                $maxMessages = 500; // 队列积压严重，大量处理
                $maxProcessTime = 0.5; // 允许处理500ms
            } elseif ($queueLength > 500) {
                $maxMessages = 300;
                $maxProcessTime = 0.3; // 允许处理300ms
            } elseif ($queueLength > 100) {
                $maxMessages = 100;
                $maxProcessTime = 0.2; // 允许处理200ms
            } elseif ($queueLength > 10) {
                $maxMessages = 50;
                $maxProcessTime = 0.1; // 允许处理100ms
            } else {
                $maxMessages = 10;
                $maxProcessTime = 0.05; // 允许处理50ms
            }
            
            // 使用批量操作提高效率（减少Redis往返）
            $startTime = microtime(true);
            $processed = 0;
            $messages = [];
            
            // 批量获取消息（使用 pipeline 提高效率）
            $batchSize = min($maxMessages, 100); // 每次批量获取100条
            for ($batch = 0; $batch < ceil($maxMessages / $batchSize); $batch++) {
                // 检查处理时间
                if (microtime(true) - $startTime > $maxProcessTime) {
                    break;
                }
                
                // 批量获取消息
                $batchMessages = [];
                for ($i = 0; $i < $batchSize && count($batchMessages) < $maxMessages - $processed; $i++) {
                    $message = $this->redis->getClient()->rpop('room_messages_queue');
                    if (!$message) {
                        break 2; // 队列为空，退出所有循环
                    }
                    $batchMessages[] = $message;
                }
                
                // 批量处理消息
                foreach ($batchMessages as $message) {
                    $this->handleRedisMessage($message);
                    $processed++;
                }
                
                // 如果队列变空，退出
                if (count($batchMessages) < $batchSize) {
                    break;
                }
            }
            
            // 记录处理情况（仅在队列较长或处理量大时记录，避免日志过多）
            if ($queueLength > 100 || $processed > 50) {
                \Workerman\Worker::log("处理队列消息: 队列长度={$queueLength}, 处理={$processed}, 剩余=" . ($queueLength - $processed) . ", 耗时=" . round((microtime(true) - $startTime) * 1000, 2) . "ms");
            }
        } catch (\Exception $e) {
            echo "检查Redis队列失败: " . $e->getMessage() . "\n";
            \Workerman\Worker::log("检查Redis队列失败: " . $e->getMessage());
        }
    }
    
    /**
     * 检查房间消息（限制处理数量，避免阻塞）
     * 仅在队列为空或很短时才处理，避免重复处理
     */
    private function checkRoomMessages()
    {
        try {
            // 先检查主队列，如果队列很长，优先处理队列消息，跳过房间消息（避免重复）
            $queueLength = $this->redis->getClient()->llen('room_messages_queue');
            if ($queueLength > 10) {
                // 队列还有消息，优先处理队列，跳过房间消息（避免重复处理）
                return;
            }
            
            // 限制每次只扫描少量 key，避免阻塞事件循环
            $cursor = $this->getLastScanCursor(); // 保存上次的 cursor
            
            // 根据队列情况调整处理数量：队列短时可以多处理一些房间消息
            $maxKeys = $queueLength > 0 ? 5 : 10; // 队列为空时处理10个，队列有少量消息时处理5个
            
            $result = $this->redis->getClient()->scan($cursor, ['match' => 'room_message:*', 'count' => 20]);
            $cursor = $result[0];
            $keys = $result[1];
            
            // 保存 cursor 供下次使用
            $this->setLastScanCursor($cursor);
            
            // 限制处理数量
            $keys = array_slice($keys, 0, $maxKeys);
            
            // 限制处理时间
            $startTime = microtime(true);
            $maxProcessTime = 0.05; // 最多处理50ms
            
            foreach ($keys as $key) {
                if (microtime(true) - $startTime > $maxProcessTime) {
                    break;
                }
                $this->processRoomMessages($key);
            }
            
            // 如果 cursor 为 0，说明扫描完成，重置
            if ($cursor === 0) {
                $this->setLastScanCursor(0);
            }
        } catch (\Exception $e) {
            echo "检查房间消息失败: " . $e->getMessage() . "\n";
        }
    }
    
    /**
     * 获取上次扫描的 cursor（使用静态变量存储）
     */
    private static $lastScanCursor = 0;
    
    private function getLastScanCursor()
    {
        return self::$lastScanCursor;
    }
    
    private function setLastScanCursor($cursor)
    {
        self::$lastScanCursor = $cursor;
    }

    /**
     * 处理单个房间的消息
     */
    private function processRoomMessages($key)
    {
        try {
            $roomId = str_replace('room_message:', '', $key);
            $messages = $this->redis->getClient()->lrange($key, 0, -1);
            
            if (!empty($messages)) {
                // 删除已读取的消息
                $this->redis->getClient()->del($key);
                
                // 广播消息到房间
                foreach ($messages as $messageData) {
                    $message = json_decode($messageData, true);
                    if ($message) {
                        $this->broadcastToRoom($roomId, $message);
                    }
                }
            }
        } catch (\Exception $e) {
            echo "处理房间消息失败: " . $e->getMessage() . "\n";
        }
    }

    /**
     * 连接建立时
     */
    public function onConnect(TcpConnection $connection)
    {
        $timestamp = date('Y-m-d H:i:s');
        $workerId = $this->workerId ?? '未知';
        $processId = getmypid() ?? '未知';
        
        $logMessage = "\n" .
            "========================================\n" .
            "========== 新WebSocket连接建立 ==========\n" .
            "========================================\n" .
            "时间戳: {$timestamp}\n" .
            "Worker ID: {$workerId}\n" .
            "进程ID: {$processId}\n" .
            "连接ID: " . $connection->id . "\n" .
            "连接IP: " . ($connection->getRemoteIp() ?? '未知') . "\n" .
            "连接端口: " . ($connection->getRemotePort() ?? '未知') . "\n" .
            "本地IP: " . ($connection->getLocalIp() ?? '未知') . "\n" .
            "本地端口: " . ($connection->getLocalPort() ?? '未知') . "\n" .
            "当前总连接数(users): " . count($this->users) . "\n" .
            "当前总房间数(rooms): " . count($this->rooms) . "\n" .
            "注意: 连接建立后需要前端发送 'connect' 消息才能注册到房间\n" .
            "========================================\n";
        
        // 输出到控制台
        echo $logMessage;
        
        // 同时记录到日志文件（使用 Worker::log 确保在 daemon 模式下也能记录）
        \Workerman\Worker::log($logMessage);
        
        // 强制刷新输出缓冲区，确保日志立即显示
        if (function_exists('flush')) {
            flush();
        }
    }

    /**
     * 收到消息时
     */
    public function onMessage(TcpConnection $connection, $data)
    {
        try {
            echo "========== 收到WebSocket消息 ==========\n";
            echo "消息内容: " . $data . "\n";
            echo "连接ID: " . $connection->id . "\n";
            echo "当前rooms数组: " . json_encode(array_keys($this->rooms)) . "\n";
            echo "当前users数组: " . json_encode(array_keys($this->users)) . "\n";
            
            $message = json_decode($data, true);
            if (!$message) {
                echo "无效的消息格式: " . $data . "\n";
                $this->sendError($connection, '无效的消息格式');
                return;
            }

            $type = $message['type'] ?? '';
            $roomId = $message['room_id'] ?? null;
            $userId = $message['user_id'] ?? null;

            echo "处理消息类型: {$type}, 房间ID: {$roomId}, 用户ID: {$userId}\n";
            
            // 如果消息中没有 userId，尝试从连接映射中获取
            if (!$userId && isset($this->connectionUsers[$connection->id])) {
                $userId = $this->connectionUsers[$connection->id];
                echo "从连接映射获取用户ID: {$userId}\n";
            }
            
            echo "连接对应的用户ID: " . ($userId ?? '未知') . "\n";
            
            // ⚠️ 重要：刷新 Redis 键的过期时间（每次收到消息时）
            // 这确保活跃连接的映射不会被过期清除
            if ($userId) {
                $this->refreshUserRedisKeys($userId, $roomId);
            }

            switch ($type) {
                case 'connect':
                case 'authenticate':
                case 'refresh_rooms':
                    // 前端连接时，自动订阅用户所在的所有房间
                    // refresh_rooms 用于在用户加入房间后，主动刷新房间订阅
                    echo "处理用户连接/认证/刷新房间请求，用户ID: {$userId}, 类型: {$type}\n";
                    if (!$userId) {
                        echo "❌ 错误: 消息类型为 {$type}，但用户ID为空！\n";
                        echo "完整消息内容: " . json_encode($message, JSON_UNESCAPED_UNICODE) . "\n";
                        $this->sendError($connection, '用户ID不能为空');
                        break;
                    }
                    try {
                        $this->handleUserConnect($connection, $userId);
                    } catch (\Exception $e) {
                        echo "❌ handleUserConnect 执行失败: " . $e->getMessage() . "\n";
                        echo "异常文件: " . $e->getFile() . "\n";
                        echo "异常行号: " . $e->getLine() . "\n";
                        echo "堆栈跟踪: " . $e->getTraceAsString() . "\n";
                        $this->sendError($connection, '处理用户连接失败: ' . $e->getMessage());
                    }
                    break;
                    
                case 'join_room':
                    echo "处理加入房间请求\n";
                    $this->handleJoinRoom($connection, $roomId, $userId);
                    break;
                    
                case 'leave_room':
                    $this->handleLeaveRoom($connection, $roomId, $userId);
                    break;
                    
                // 添加积分支付消息处理
                case 'score_transfer':
                    $this->handleScoreTransfer($connection, $message);
                    break;
                    
                // 移除对前端直接发送的记账消息的处理
                // 记账消息只应由后端API通过WebSocketService进行广播
                // case 'expense_recorded':
                //     $this->handleExpenseRecorded($connection, $message);
                //     break;
                    
                case 'room_settlement':
                    $this->handleRoomSettlement($connection, $roomId, $message);
                    break;
                    
                default:
                    $this->sendError($connection, '未知的消息类型');
            }
        } catch (\Exception $e) {
            echo "❌ 处理WebSocket消息异常 ==========\n";
            echo "异常消息: " . $e->getMessage() . "\n";
            echo "异常文件: " . $e->getFile() . "\n";
            echo "异常行号: " . $e->getLine() . "\n";
            echo "堆栈跟踪: " . $e->getTraceAsString() . "\n";
            echo "原始数据: " . $data . "\n";
            $this->sendError($connection, '服务器内部错误: ' . $e->getMessage());
        }
    }
    
    /**
     * 刷新用户相关的 Redis 键的过期时间
     * 在用户有活动时调用，防止 Redis 键过期导致映射丢失
     */
    private function refreshUserRedisKeys($userId, $roomId = null)
    {
        try {
            // 刷新用户 worker 映射的过期时间
            $userWorkerKey = $this->getUserWorkerKey($userId);
            if ($this->redis->getClient()->exists($userWorkerKey)) {
                $this->redis->getClient()->expire($userWorkerKey, 3600); // 重置为1小时
                echo "✅ 已刷新用户 {$userId} 的 worker 映射过期时间\n";
            } else {
                // 如果键不存在，重新创建（可能在过期后用户还在连接）
                $this->redis->getClient()->setex($userWorkerKey, 3600, $this->workerId);
                echo "⚠️ 用户 {$userId} 的 worker 映射已过期，已重新创建\n";
            }
            
            // 如果提供了房间ID，刷新房间相关的键
            if ($roomId) {
                $roomId = (string)$roomId;
                
                // 刷新房间用户集合的过期时间
                $roomUsersKey = $this->getRoomUsersKey($roomId);
                if ($this->redis->getClient()->exists($roomUsersKey)) {
                    $this->redis->getClient()->expire($roomUsersKey, 3600);
                    echo "✅ 已刷新房间 {$roomId} 的用户集合过期时间\n";
                }
                
                // 刷新用户房间集合的过期时间
                $userRoomsKey = $this->getUserRoomsKey($userId);
                if ($this->redis->getClient()->exists($userRoomsKey)) {
                    $this->redis->getClient()->expire($userRoomsKey, 3600);
                    // 确保房间ID在集合中
                    $this->redis->getClient()->sadd($userRoomsKey, $roomId);
                    echo "✅ 已刷新用户 {$userId} 的房间集合过期时间\n";
                }
            } else {
                // 如果没有提供房间ID，尝试从 Redis 获取用户的所有房间
                $userRoomsKey = $this->getUserRoomsKey($userId);
                $userRooms = $this->redis->getClient()->smembers($userRoomsKey);
                if (!empty($userRooms)) {
                    // 刷新用户房间集合
                    $this->redis->getClient()->expire($userRoomsKey, 3600);
                    
                    // 刷新每个房间的用户集合
                    foreach ($userRooms as $rId) {
                        $roomUsersKey = $this->getRoomUsersKey($rId);
                        if ($this->redis->getClient()->exists($roomUsersKey)) {
                            $this->redis->getClient()->expire($roomUsersKey, 3600);
                        }
                    }
                    echo "✅ 已刷新用户 {$userId} 的所有房间相关键（" . count($userRooms) . "个房间）\n";
                }
            }
        } catch (\Exception $e) {
            echo "⚠️ 刷新用户 Redis 键过期时间失败: " . $e->getMessage() . "\n";
            // 不抛出异常，不影响消息处理
        }
    }

    /**
     * 连接关闭时
     */
    public function onClose(TcpConnection $connection)
    {
        echo "========== WebSocket连接关闭 ==========\n";
        echo "连接ID: " . $connection->id . "\n";
        echo "连接IP: " . ($connection->getRemoteIp() ?? '未知') . "\n";
        echo "连接端口: " . ($connection->getRemotePort() ?? '未知') . "\n";
        
        // 查找并清理用户连接映射（本进程内）
        $closedUserId = null;
        foreach ($this->users as $userId => $conn) {
            if ($conn === $connection) {
                $closedUserId = $userId;
                unset($this->users[$userId]);
                echo "清理用户连接映射: 用户ID {$userId}\n";
                break;
            }
        }
        if (!$closedUserId && isset($this->connectionUsers[$connection->id])) {
            $closedUserId = $this->connectionUsers[$connection->id];
            echo "从连接映射中找到用户ID: {$closedUserId}\n";
        }
        unset($this->connectionUsers[$connection->id]);
        
        // 清理房间连接映射（本进程内）
        $closedRooms = [];
        foreach ($this->rooms as $roomId => $connections) {
            $key = array_search($connection, $connections, true);
            if ($key !== false) {
                unset($this->rooms[$roomId][$key]);
                $closedRooms[] = $roomId;
                if (empty($this->rooms[$roomId])) {
                    unset($this->rooms[$roomId]);
                    echo "清理空房间: 房间ID {$roomId}\n";
                } else {
                    echo "从房间 {$roomId} 移除连接，剩余连接数: " . count($this->rooms[$roomId]) . "\n";
                }
            }
        }
        
        // 清理用户房间映射（本进程内）
        if ($closedUserId && isset($this->userRooms[$closedUserId])) {
            unset($this->userRooms[$closedUserId]);
            echo "清理用户房间映射: 用户ID {$closedUserId}\n";
        }
        
        // ⚠️ 重要：连接关闭时不立即删除 Redis 映射
        // 原因：
        // 1. 用户可能只是短暂断开（网络波动、切换 tab 等），很快会重连
        // 2. 立即删除会导致映射丢失，重连后需要重新发送 connect 消息
        // 3. 如果用户真的离开了，Redis 键会在1小时后自动过期（已设置过期时间）
        // 
        // 只有用户明确发送 leave_room 消息时，才清理 Redis 映射
        // 这里只清理本进程内的内存映射，保留 Redis 映射以便快速重连
        
        if ($closedUserId) {
            // 延迟删除 Redis 映射（5分钟后），给重连留出时间窗口
            // 如果用户在5分钟内重连，映射仍然存在，可以快速恢复
            $userWorkerKey = $this->getUserWorkerKey($closedUserId);
            
            // 将过期时间缩短为5分钟（而不是立即删除）
            // 如果用户在5分钟内重连，映射仍然存在；如果没重连，5分钟后自动过期
            try {
                if ($this->redis->getClient()->exists($userWorkerKey)) {
                    $this->redis->getClient()->expire($userWorkerKey, 300); // 5分钟
                    echo "用户 {$closedUserId} 连接关闭，Redis worker 映射过期时间缩短为5分钟（给重连留出时间窗口）\n";
                }
            } catch (\Exception $e) {
                echo "设置延迟过期失败: " . $e->getMessage() . "\n";
                // 如果设置失败，保留原有过期时间（不删除）
            }
            
            // 注意：
            // 1. 房间用户集合不删除，因为用户可能还在房间中（只是断开连接）
            // 2. 用户房间集合不删除，同样原因
            // 3. Worker 映射设置为5分钟过期，如果用户重连会通过 refreshUserRedisKeys 恢复为1小时
        }
        
        echo "关闭后 - 当前rooms数组: " . json_encode(array_keys($this->rooms)) . "\n";
        echo "关闭后 - 当前users数组: " . json_encode(array_keys($this->users)) . "\n";
        echo "已清理的房间: " . json_encode($closedRooms) . "\n";
    }

    /**
     * 连接错误时
     */
    public function onError(TcpConnection $connection, $code, $msg)
    {
        $timestamp = date('Y-m-d H:i:s');
        $workerId = $this->workerId ?? '未知';
        
        $logMessage = "\n" .
            "========================================\n" .
            "========== WebSocket连接错误 ==========\n" .
            "========================================\n" .
            "时间戳: {$timestamp}\n" .
            "Worker ID: {$workerId}\n" .
            "连接ID: " . $connection->id . "\n" .
            "错误代码: {$code}\n" .
            "错误消息: {$msg}\n" .
            "连接IP: " . ($connection->getRemoteIp() ?? '未知') . "\n" .
            "========================================\n";
        
        // 输出到控制台
        echo $logMessage;
        
        // 同时记录到日志文件
        \Workerman\Worker::log($logMessage);
        
        // 强制刷新输出缓冲区
        if (function_exists('flush')) {
            flush();
        }
    }

    /**
     * 处理用户连接（自动订阅用户所在的所有房间）
     */
    private function handleUserConnect(TcpConnection $connection, $userId)
    {
        echo "========== 开始处理用户连接 ==========\n";
        echo "连接ID: " . $connection->id . "\n";
        echo "用户ID: {$userId} (类型: " . gettype($userId) . ")\n";
        echo "处理前 - 当前rooms数组: " . json_encode(array_keys($this->rooms)) . "\n";
        echo "处理前 - 当前users数组: " . json_encode(array_keys($this->users)) . "\n";
        
        if (!$userId) {
            echo "❌ 错误: 用户ID为空，无法处理连接\n";
            $this->sendError($connection, '用户ID不能为空');
            return;
        }

        // 确保用户ID是整数类型
        $userId = (int)$userId;
        if ($userId <= 0) {
            echo "❌ 错误: 用户ID无效: {$userId}\n";
            $this->sendError($connection, '用户ID无效');
            return;
        }

        // 验证用户是否存在
        try {
            $user = User::find($userId);
            if (!$user) {
                echo "❌ 错误: 用户不存在: {$userId}\n";
                $this->sendError($connection, '用户不存在');
                return;
            }
            echo "✅ 用户验证通过: {$user->nickname} (ID: {$user->id})\n";
        } catch (\Exception $e) {
            echo "❌ 查询用户失败: " . $e->getMessage() . "\n";
            $this->sendError($connection, '查询用户失败');
            return;
        }

        // 检查是否为重连（用户ID已存在但连接不同）
        $isReconnect = false;
        if (isset($this->users[$userId]) && $this->users[$userId] !== $connection) {
            echo "检测到用户 {$userId} 重连，旧连接将被替换\n";
            $isReconnect = true;
            
            // 获取旧连接并清理
            $oldConnection = $this->users[$userId];
            foreach ($this->rooms as $roomIdInList => $roomConnections) {
                if (($key = array_search($oldConnection, $roomConnections)) !== false) {
                    unset($this->rooms[$roomIdInList][$key]);
                    if (empty($this->rooms[$roomIdInList])) {
                        unset($this->rooms[$roomIdInList]);
                    }
                }
            }
        }

        // 更新连接到用户ID的映射（本进程内）
        $this->connectionUsers[$connection->id] = $userId;
        $this->users[$userId] = $connection;
        
        // 同步到 Redis：标记用户连接在哪个 worker 进程
        $this->redis->getClient()->setex(
            $this->getUserWorkerKey($userId),
            3600, // 1小时过期
            $this->workerId
        );
        echo "✅ 已同步用户 {$userId} 到 Redis，标记在 Worker {$this->workerId}\n";

        // 获取用户所在的所有活跃房间
        try {
            echo "查询用户 {$userId} 所在的活跃房间...\n";
            $userRooms = RoomUser::where('user_id', $userId)
                ->where('status', RoomUser::STATUS_ACTIVE)
                ->with('room')
                ->get();
            
            $roomCount = $userRooms->count();
            echo "找到 {$roomCount} 个活跃房间记录\n";
            
            if ($roomCount === 0) {
                echo "⚠️ 用户 {$userId} 不在任何活跃房间中\n";
                echo "这可能是正常的（用户刚刚登录，还未加入房间）\n";
            } else {
                echo "房间列表: " . $userRooms->pluck('room_id')->implode(', ') . "\n";
            }
        } catch (\Exception $e) {
            echo "❌ 查询用户房间失败: " . $e->getMessage() . "\n";
            echo "异常堆栈: " . $e->getTraceAsString() . "\n";
            $this->sendError($connection, '查询用户房间失败');
            return;
        }

        $subscribedRooms = [];
        $roomsData = []; // 存储每个房间的完整数据，用于全量刷新
        
        foreach ($userRooms as $roomUser) {
            echo "处理房间用户记录: room_user_id={$roomUser->id}, room_id={$roomUser->room_id}\n";
            
            if (!$roomUser->room) {
                echo "⚠️ 房间用户记录存在但房间不存在: room_user_id={$roomUser->id}, room_id={$roomUser->room_id}\n";
                continue;
            }

            $roomId = (string)$roomUser->room_id;
            $room = $roomUser->room;
            echo "处理房间 {$roomId}: {$room->name} (状态: {$room->status})\n";
            $subscribedRooms[] = $roomId;

            // 将连接添加到房间
            if (!isset($this->rooms[$roomId])) {
                $this->rooms[$roomId] = [];
                echo "创建房间 {$roomId} 的连接数组\n";
            }

            // 移除同连接在同房间的旧记录（如果存在）
            $key = array_search($connection, $this->rooms[$roomId], true);
            if ($key !== false) {
                unset($this->rooms[$roomId][$key]);
                echo "移除房间 {$roomId} 中的旧连接记录\n";
            }

            $this->rooms[$roomId][] = $connection;
            echo "✅ 用户 {$userId} 的连接已添加到房间 {$roomId}\n";
            echo "房间 {$roomId} 当前连接数: " . count($this->rooms[$roomId]) . "\n";

            // 更新用户房间映射（本进程内）
            if (!isset($this->userRooms[$userId])) {
                $this->userRooms[$userId] = [];
            }
            if (!in_array($roomId, $this->userRooms[$userId])) {
                $this->userRooms[$userId][] = $roomId;
                echo "更新用户 {$userId} 的房间映射，新增房间 {$roomId}\n";
            }
            
            // ⚠️ 重要：同步当前用户到 Redis
            // 同时确保房间中的其他用户也被同步到 Redis（追加模式，不覆盖）
            $roomUsersKey = $this->getRoomUsersKey($roomId);
            
            // 将当前用户添加到房间的用户集合（追加）
            $this->redis->getClient()->sadd($roomUsersKey, $userId);
            echo "✅ 已将用户 {$userId} 添加到房间 {$roomId} 的 Redis 用户集合\n";
            
            // ⚠️ 关键：确保房间中的所有活跃用户都被同步到 Redis（追加，不覆盖）
            // 获取房间中的所有活跃用户（从数据库）
            try {
                $allRoomUsers = RoomUser::where('room_id', $room->id)
                    ->where('status', RoomUser::STATUS_ACTIVE)
                    ->pluck('user_id')
                    ->toArray();
                
                echo "房间 {$roomId} 数据库中的活跃用户: " . json_encode($allRoomUsers) . "\n";
                
                // 将房间中的所有活跃用户都添加到 Redis（追加模式）
                if (!empty($allRoomUsers)) {
                    foreach ($allRoomUsers as $roomUserId) {
                        // 添加到房间用户集合（sadd 是追加，如果已存在不会重复）
                        $this->redis->getClient()->sadd($roomUsersKey, $roomUserId);
                        
                        // 同步用户房间集合（确保用户的房间映射也更新）
                        $userRoomsKey = $this->getUserRoomsKey($roomUserId);
                        $this->redis->getClient()->sadd($userRoomsKey, $roomId);
                        $this->redis->getClient()->expire($userRoomsKey, 3600);
                    }
                    echo "✅ 已同步房间 {$roomId} 的所有活跃用户到 Redis（" . count($allRoomUsers) . "个用户）\n";
                }
            } catch (\Exception $e) {
                echo "⚠️ 同步房间用户到 Redis 失败: " . $e->getMessage() . "\n";
                // 即使同步失败，也继续处理
            }
            
            // 设置房间用户集合的过期时间
            $this->redis->getClient()->expire($roomUsersKey, 3600); // 1小时过期
            
            // 同步到 Redis：将房间添加到当前用户的房间集合
            $this->redis->getClient()->sadd($this->getUserRoomsKey($userId), $roomId);
            $this->redis->getClient()->expire($this->getUserRoomsKey($userId), 3600); // 1小时过期

            // 获取房间的完整状态数据（全量刷新）
            try {
                // 刷新房间模型以确保获取最新数据
                $room->refresh();
                $room->updateCurrentPlayers();
                $room->refresh();
                
                // 获取排序后的用户列表（使用WebSocketService的方法）
                $websocketService = \app\service\WebSocketService::getInstance();
                $sortedUsers = $websocketService->getSortedRoomUsersForConnect($room);
                
                $roomsData[$roomId] = [
                    'room_id' => (int)$roomId,
                    'room_code' => $room->room_code,
                    'name' => $room->name,
                    'status' => $room->status,
                    'current_players' => $room->current_players,
                    'max_players' => $room->max_players,
                    'created_by' => $room->created_by,
                    'users' => $sortedUsers, // 完整的用户列表（包含最新分数）
                    'updated_at' => time()
                ];
                
                echo "用户 {$userId} 已自动订阅房间 {$roomId}，已获取房间完整状态\n";
            } catch (\Exception $e) {
                echo "获取房间 {$roomId} 状态失败: " . $e->getMessage() . "\n";
                // 即使获取失败，也继续处理其他房间
                $roomsData[$roomId] = [
                    'room_id' => (int)$roomId,
                    'error' => '获取房间状态失败'
                ];
            }
        }

        // 发送连接成功消息，包含已订阅的房间列表和每个房间的完整状态（用于全量刷新）
        $this->sendMessage($connection, [
            'type' => 'connect_success',
            'user_id' => $userId,
            'is_reconnect' => $isReconnect,
            'subscribed_rooms' => $subscribedRooms,
            'rooms_data' => $roomsData, // 每个房间的完整状态数据
            'message' => '连接成功，已自动订阅 ' . count($subscribedRooms) . ' 个房间',
            'timestamp' => time()
        ]);

        echo "========== 用户连接处理完成 ==========\n";
        echo "已订阅房间数量: " . count($subscribedRooms) . "\n";
        echo "是否重连: " . ($isReconnect ? '是' : '否') . "\n";
        echo "处理后 - 当前rooms数组: " . json_encode(array_keys($this->rooms)) . "\n";
        echo "处理后 - 当前users数组: " . json_encode(array_keys($this->users)) . "\n";
        foreach ($this->rooms as $roomId => $connections) {
            echo "房间 {$roomId} 的连接数: " . count($connections) . "\n";
        }
    }

    /**
     * 处理加入房间
     */
    private function handleJoinRoom(TcpConnection $connection, $roomId, $userId)
    {
        // 确保房间ID使用字符串类型，以保持一致性
        $roomId = (string)$roomId;
        
        echo "处理加入房间: 房间ID={$roomId}, 用户ID={$userId}\n";
        
        if (!$roomId || !$userId) {
            echo "房间ID或用户ID为空\n";
            $this->sendError($connection, '房间ID和用户ID不能为空');
            return;
        }

        // 验证房间是否存在
        $room = Room::find($roomId);
        if (!$room) {
            echo "房间不存在: {$roomId}\n";
            $this->sendError($connection, '房间不存在');
            return;
        }
        echo "房间验证通过: {$room->name}\n";

        // 验证用户是否存在
        $user = User::find($userId);
        if (!$user) {
            echo "用户不存在: {$userId}\n";
            $this->sendError($connection, '用户不存在');
            return;
        }
        echo "用户验证通过: {$user->nickname}\n";

        // 检查是否为重连（用户ID已存在但连接不同）
        $isReconnect = false;
        if (isset($this->users[$userId]) && $this->users[$userId] !== $connection) {
            echo "检测到用户 {$userId} 重连，旧连接将被替换\n";
            $isReconnect = true;
            
            // 获取旧连接
            $oldConnection = $this->users[$userId];
            
            // 从所有房间中移除旧连接
            foreach ($this->rooms as $roomIdInList => $roomConnections) {
                if (($key = array_search($oldConnection, $roomConnections)) !== false) {
                    unset($this->rooms[$roomIdInList][$key]);
                    if (empty($this->rooms[$roomIdInList])) {
                        unset($this->rooms[$roomIdInList]);
                    }
                }
            }
        }

        // 更新连接到用户ID的映射
        $this->connectionUsers[$connection->id] = $userId;
        
        // 将连接添加到房间
        if (!isset($this->rooms[$roomId])) {
            $this->rooms[$roomId] = [];
        }
        
        // 移除同连接在同房间的旧记录（如果存在）
        $key = array_search($connection, $this->rooms[$roomId]);
        if ($key !== false) {
            unset($this->rooms[$roomId][$key]);
        }
        
        $this->rooms[$roomId][] = $connection;
        
        // 更新用户到连接的映射
        $this->users[$userId] = $connection;
        
        // 更新用户房间映射
        if (!isset($this->userRooms[$userId])) {
            $this->userRooms[$userId] = [];
        }
        if (!in_array($roomId, $this->userRooms[$userId])) {
            $this->userRooms[$userId][] = $roomId;
        }
        
        echo "用户已加入房间，当前房间连接数: " . count($this->rooms[$roomId]) . "\n";
        echo "房间 {$roomId} 的连接详情: " . json_encode(array_keys($this->rooms[$roomId])) . "\n";

        // 创建用户加入房间的活动记录
        try {
            $this->createActivityRecord($roomId, $userId, $userId, 0, $isReconnect ? 'user_rejoined' : 'user_joined', $user->nickname . ($isReconnect ? ' 回到了房间' : ' 进入了房间'));
        } catch (\Exception $e) {
            echo "创建加入房间活动记录失败: " . $e->getMessage() . "\n";
        }

        // 获取历史消息
        $historyMessages = $this->getRoomHistoryMessages($roomId);
        
        // 发送历史消息给新连接的用户
        if (!empty($historyMessages)) {
            echo "向用户 {$userId} 发送历史消息，共 " . count($historyMessages) . " 条\n";
            $this->sendMessage($connection, [
                'type' => 'history_messages',
                'messages' => $historyMessages,
                'timestamp' => time()
            ]);
        }
        
        // 根据是否重连发送不同的通知
        $notificationType = $isReconnect ? 'user_rejoined' : 'user_joined';
        $notificationMessage = $isReconnect ? $user->nickname . ' 回到了房间' : $user->nickname . ' 进入了房间';
        
        // 通知房间内其他用户
        $this->broadcastToRoom($roomId, [
            'type' => $notificationType,
            'message' => $notificationMessage,
            'user' => [
                'id' => $user->id,
                'nickname' => $user->nickname,
                'avatar' => $user->avatar
            ],
            'timestamp' => time()
        ], $connection);

        // 发送确认消息给当前用户
        $this->sendMessage($connection, [
            'type' => 'join_room_success',
            'room_id' => $roomId,
            'user_id' => $userId,
            'is_reconnect' => $isReconnect,
            'timestamp' => time()
        ]);
        
        echo "加入房间处理完成，是否重连: " . ($isReconnect ? '是' : '否') . "\n";
    }
    
    /**
     * 创建活动记录
     */
    private function createActivityRecord($roomId, $userId, $targetUserId, $score, $type, $description = '')
    {
        try {
            // 如果是0分记录且不是系统记录，则不创建
            if ($score == 0 && !in_array($type, ['user_joined', 'user_left', 'user_rejoined'])) {
                echo "跳过0分记录: " . json_encode(['type' => $type, 'score' => $score, 'description' => $description]) . "\n";
                return null;
            }
            
            // 使用AccountingRecord模型创建记录
            $recordData = [
                'room_id' => (int)$roomId,
                'user_id' => (int)$userId,
                'target_user_id' => (int)$targetUserId,
                'score' => (int)$score,
                'description' => $description,
                'type' => $type,
                'is_settled' => 0
            ];
            
            echo "创建活动记录: " . json_encode($recordData) . "\n";
            $record = \app\model\AccountingRecord::createRecord($recordData);
            echo "活动记录创建成功，ID: " . $record->id . "\n";
            return $record;
        } catch (\Exception $e) {
            echo "创建活动记录异常: " . $e->getMessage() . "\n";
            echo "异常文件: " . $e->getFile() . "\n";
            echo "异常行号: " . $e->getLine() . "\n";
            throw $e;
        }
    }
    
    /**
     * 获取房间历史消息
     */
    private function getRoomHistoryMessages($roomId)
    {
        try {
            // 从Redis获取历史消息（保存最近50条）
            $historyKey = "room_history:{$roomId}";
            $messages = $this->redis->getClient()->lrange($historyKey, 0, 49);
            
            // 解析消息
            $parsedMessages = [];
            foreach ($messages as $messageData) {
                $message = json_decode($messageData, true);
                if ($message) {
                    $parsedMessages[] = $message;
                }
            }
            
            // 反转数组，使最早的消息在前
            return array_reverse($parsedMessages);
        } catch (\Exception $e) {
            echo "获取房间历史消息失败: " . $e->getMessage() . "\n";
            return [];
        }
    }

    /**
     * 处理离开房间
     * ⚠️ 重要：只移除当前用户的映射，不影响房间中的其他用户
     */
    private function handleLeaveRoom(TcpConnection $connection, $roomId, $userId)
    {
        if (!$roomId || !$userId) {
            echo "⚠️ 警告：handleLeaveRoom 被调用，但 roomId 或 userId 为空\n";
            return;
        }
        
        $roomId = (string)$roomId;
        $userId = (int)$userId;
        
        echo "========== 处理用户离开房间 ==========\n";
        echo "房间ID: {$roomId}, 用户ID: {$userId}\n";
        echo "⚠️ 重要：只移除用户 {$userId} 的映射，不影响房间 {$roomId} 中的其他用户\n";

        // 从房间连接中移除（只移除当前连接）
        if (isset($this->rooms[$roomId])) {
            $key = array_search($connection, $this->rooms[$roomId]);
            if ($key !== false) {
                unset($this->rooms[$roomId][$key]);
                echo "已从房间 {$roomId} 的连接列表移除当前连接\n";
            }
            if (empty($this->rooms[$roomId])) {
                unset($this->rooms[$roomId]);
                echo "房间 {$roomId} 的连接列表已清空（本进程内）\n";
            } else {
                echo "房间 {$roomId} 还有 " . count($this->rooms[$roomId]) . " 个连接（本进程内）\n";
            }
        }

        // 从用户映射中移除（本进程内，只移除当前用户）
        if (isset($this->users[$userId])) {
            unset($this->users[$userId]);
            echo "已从 users 映射中移除用户 {$userId}（本进程内）\n";
        }
        
        // ⚠️ 重要：从 Redis 中移除当前用户的映射（只移除当前用户，不影响其他用户）
        try {
            $roomUsersKey = $this->getRoomUsersKey($roomId);
            
            // 获取移除前的用户列表（用于验证）
            $beforeUsers = $this->redis->getClient()->smembers($roomUsersKey);
            echo "移除前 - 房间 {$roomId} 的 Redis 用户集合: " . json_encode($beforeUsers) . "\n";
            
            // 使用 srem 只移除当前用户（不影响其他用户）
            $removed = $this->redis->getClient()->srem($roomUsersKey, $userId);
            if ($removed > 0) {
                echo "✅ 已从 Redis 房间用户集合移除用户 {$userId}（srem 返回: {$removed}）\n";
            } else {
                echo "⚠️ 用户 {$userId} 不在房间 {$roomId} 的 Redis 集合中（可能已移除或不存在）\n";
            }
            
            // 获取移除后的用户列表（用于验证）
            $afterUsers = $this->redis->getClient()->smembers($roomUsersKey);
            echo "移除后 - 房间 {$roomId} 的 Redis 用户集合: " . json_encode($afterUsers) . "\n";
            echo "✅ 确认：其他用户 " . json_encode(array_diff($beforeUsers, [$userId])) . " 仍然在 Redis 集合中\n";
            
            // 从用户房间集合中移除房间（只影响当前用户）
            $userRoomsKey = $this->getUserRoomsKey($userId);
            $removedFromUserRooms = $this->redis->getClient()->srem($userRoomsKey, $roomId);
            if ($removedFromUserRooms > 0) {
                echo "✅ 已从用户 {$userId} 的房间集合移除房间 {$roomId}\n";
            }
            
            // 检查用户是否还在其他房间
            $remainingRooms = $this->redis->getClient()->smembers($userRoomsKey);
            if (empty($remainingRooms)) {
                // 如果用户不在任何房间了，删除 worker 映射（只影响当前用户）
                $this->redis->getClient()->del($this->getUserWorkerKey($userId));
                echo "用户 {$userId} 已离开所有房间，已删除该用户的 worker 映射\n";
            } else {
                echo "用户 {$userId} 还在其他房间: " . json_encode($remainingRooms) . "\n";
            }
        } catch (\Exception $e) {
            echo "❌ 清理 Redis 映射失败: " . $e->getMessage() . "\n";
            echo "异常堆栈: " . $e->getTraceAsString() . "\n";
        }
        
        // 创建用户离开房间的活动记录
        try {
            // 获取用户昵称
            $user = User::find($userId);
            $nickname = $user ? $user->nickname : '用户';
            $this->createActivityRecord($roomId, $userId, $userId, 0, 'user_left', $nickname . ' 离开了房间');
        } catch (\Exception $e) {
            echo "创建离开房间活动记录失败: " . $e->getMessage() . "\n";
        }

        // 通知房间内其他用户
        $this->broadcastToRoom($roomId, [
            'type' => 'user_left',
            'user' => [
                'id' => $userId,
                'nickname' => '用户' // 这里可以从数据库获取，但为了性能考虑暂时简化
            ],
            'timestamp' => time()
        ]);
    }

    /**
     * 处理支出记录
     */
    private function handleExpenseRecorded(TcpConnection $connection, $message)
    {
        $roomId = $message['room_id'] ?? null;
        $fromUserId = $message['from_user']['id'] ?? null;
        $toUserId = $message['to_user']['id'] ?? null;
        $amount = $message['amount'] ?? 0;
        $expenseType = $message['expense_type'] ?? 'score';
        
        if (!$roomId) {
            $this->sendError($connection, '房间ID不能为空');
            return;
        }
        
        if (!$fromUserId || !$toUserId) {
            $this->sendError($connection, '用户ID不能为空');
            return;
        }

        // 创建支出记录的活动记录
        try {
            // 获取用户信息
            $fromUser = User::find($fromUserId);
            $toUser = User::find($toUserId);
            $fromNickname = $fromUser ? $fromUser->nickname : '用户';
            $toNickname = $toUser ? $toUser->nickname : '用户';
            
            $description = $expenseType === 'tea_meal' ? 
                "{$fromNickname} 支付茶饭费给 {$toNickname}" : 
                "{$fromNickname} 支付分数给 {$toNickname}";
            
            $this->createActivityRecord($roomId, $fromUserId, $toUserId, $amount, $expenseType, $description);
        } catch (\Exception $e) {
            echo "创建支出记录活动记录失败: " . $e->getMessage() . "\n";
        }

        // 广播给房间内所有用户
        $this->broadcastToRoom($roomId, [
            'type' => 'expense_recorded',
            'from_user' => $message['from_user'] ?? null,
            'to_user' => $message['to_user'] ?? null,
            'amount' => $amount,
            'expense_type' => $expenseType,
            'timestamp' => time()
        ]);
    }

    /**
     * 处理积分支付
     */
    private function handleScoreTransfer(TcpConnection $connection, $message)
    {
        $roomId = $message['room_id'] ?? null;
        $fromUserId = $message['from_user_id'] ?? null;
        $toUserId = $message['to_user_id'] ?? null;
        $amount = $message['amount'] ?? 0;
        $expenseType = $message['expense_type'] ?? 'score';
        
        if (!$roomId) {
            $this->sendError($connection, '房间ID不能为空');
            return;
        }
        
        if (!$fromUserId || !$toUserId) {
            $this->sendError($connection, '用户ID不能为空');
            return;
        }

        if ($amount <= 0) {
            $this->sendError($connection, '支付金额必须大于0');
            return;
        }

        // 创建积分支付的活动记录
        try {
            // 获取用户信息
            $fromUser = User::find($fromUserId);
            $toUser = User::find($toUserId);
            
            if (!$fromUser || !$toUser) {
                $this->sendError($connection, '用户信息不存在');
                return;
            }
            
            // 验证用户是否在房间中
            $roomUser = RoomUser::where('room_id', $roomId)->where('user_id', $fromUserId)->find();
            if (!$roomUser) {
                $this->sendError($connection, '您不在该房间中');
                return;
            }
            
            $fromNickname = $fromUser->nickname;
            $toNickname = $toUser->nickname;
            
            $description = "{$fromNickname} 支付 {$amount} 积分给 {$toNickname}";
            
            echo "创建积分支付活动记录: {$description}\n";
            // 使用当前房间ID创建活动记录，确保基于房间维度
            $record = $this->createActivityRecord($roomId, $fromUserId, $toUserId, $amount, $expenseType, $description);
            
            // 更新房间用户分数 - 确保在房间维度上更新用户分数
            $accountingController = new \app\controller\AccountingController();
            $accountingController->updateRoomUserScore($roomId, $toUserId, $amount);
            $accountingController->updateRoomUserScore($roomId, $fromUserId, -$amount); // 减少付款方分数

            // 记录交易审计日志 - 增加审计跟踪
            $accountingController->logTransaction($record, $roomId, $fromUserId, $toUserId, $amount);
            
            // 使用WebSocketService发送积分支付消息给房间内所有用户
            \app\service\WebSocketService::sendScoreTransferMessage($roomId, $fromUserId, $toUserId, $amount, $description, $record->id);
            
            // 发送确认消息给当前用户
            $this->sendMessage($connection, [
                'type' => 'score_transfer_success',
                'message' => '积分支付成功',
                'transaction_id' => $record->id,
                'room_id' => $roomId,
                'timestamp' => time()
            ]);
            
        } catch (\Exception $e) {
            echo "创建积分支付活动记录失败: " . $e->getMessage() . "\n";
            $this->sendError($connection, '积分支付失败: ' . $e->getMessage());
        }
    }

    /**
     * 处理房间结算
     */
    private function handleRoomSettlement(TcpConnection $connection, $roomId, $message)
    {
        if (!$roomId) {
            $this->sendError($connection, '房间ID不能为空');
            return;
        }

        // 广播结算消息给房间内所有用户
        $this->broadcastToRoom($roomId, [
            'type' => 'room_settlement',
            'message' => $message['message'] ?? '房间结算完成',
            'time' => $message['time'] ?? date('H:i'),
            'settlement_data' => $message['settlement_data'] ?? [],
            'timestamp' => time()
        ]);
    }

    /**
     * 向房间内所有用户广播消息
     * 使用 Redis 获取房间内的所有用户，然后在本进程内查找连接并广播
     */
    private function broadcastToRoom($roomId, $message, $excludeConnection = null)
    {
        // 确保房间ID是字符串类型
        $roomId = (string)$roomId;
        
        echo "========== 尝试广播消息到房间 {$roomId} ==========\n";
        echo "Worker ID: {$this->workerId}\n";
        
        // 从 Redis 获取房间内的所有用户ID
        $roomUsersKey = $this->getRoomUsersKey($roomId);
        $allUserIds = $this->redis->getClient()->smembers($roomUsersKey);
        
        echo "从 Redis 获取到房间 {$roomId} 的用户: " . json_encode($allUserIds) . "\n";
        
        if (empty($allUserIds)) {
            echo "⚠️ 房间 {$roomId} 在 Redis 中没有用户记录，尝试从数据库查询\n";
            // 如果 Redis 中没有，从数据库查询并同步到 Redis
            $this->syncRoomUsersFromDatabase($roomId);
            $allUserIds = $this->redis->getClient()->smembers($roomUsersKey);
        }
        
        $broadcastCount = 0;
        $notInThisWorkerCount = 0;
        
        // 遍历房间内的所有用户，只广播本进程内的连接
        foreach ($allUserIds as $userId) {
            $userId = (int)$userId;
            
            echo "检查用户 {$userId} 的连接状态...\n";
            
            // 检查用户是否在本进程内连接
            if (isset($this->users[$userId])) {
                $connection = $this->users[$userId];
                echo "✅ 用户 {$userId} 在本进程（Worker {$this->workerId}）内已连接\n";
                
                // 如果指定了排除的连接，跳过
                if ($connection === $excludeConnection) {
                    echo "跳过用户 {$userId}（排除连接）\n";
                    continue;
                }
                
                // 确保用户在房间的连接列表中
                if (!isset($this->rooms[$roomId])) {
                    $this->rooms[$roomId] = [];
                }
                if (!in_array($connection, $this->rooms[$roomId], true)) {
                    $this->rooms[$roomId][] = $connection;
                    echo "将用户 {$userId} 添加到房间 {$roomId} 的连接列表\n";
                }
                
                // 发送消息
                $this->sendMessage($connection, $message);
                $broadcastCount++;
                echo "✅ Worker {$this->workerId}: 已向用户 {$userId} 广播消息\n";
            } else {
                echo "⚠️ 用户 {$userId} 不在本进程（Worker {$this->workerId}）的连接列表中\n";
                
                // 检查 Redis 中是否有该用户的 worker 标记
                $userWorkerKey = $this->getUserWorkerKey($userId);
                $userWorker = $this->redis->getClient()->get($userWorkerKey);
                
                if ($userWorker !== null) {
                    if ($userWorker != $this->workerId) {
                        $notInThisWorkerCount++;
                        echo "用户 {$userId} 在 Worker {$userWorker} 中（不在本进程）\n";
                    } else {
                        echo "❌ 异常: 用户 {$userId} 标记在本进程（Worker {$this->workerId}），但 users 数组中没有连接\n";
                        echo "当前 users 数组键: " . json_encode(array_keys($this->users)) . "\n";
                        echo "可能原因: 连接已断开但 Redis 标记未清理，或连接注册失败\n";
                        // 清理过期的 Redis 标记
                        $this->redis->getClient()->del($userWorkerKey);
                        echo "已清理过期的 Redis worker 标记\n";
                    }
                } else {
                    echo "⚠️ 用户 {$userId} 在 Redis 中也没有 worker 标记\n";
                    echo "可能原因:\n";
                    echo "  1. 用户未发送 connect 消息\n";
                    echo "  2. connect 消息发送失败\n";
                    echo "  3. handleUserConnect 执行失败\n";
                    echo "  4. 连接已断开但 Redis 未清理\n";
                }
            }
        }
        
        echo "广播完成: Worker {$this->workerId} 成功广播 {$broadcastCount} 个用户，{$notInThisWorkerCount} 个用户在其他进程\n";
    }
    
    /**
     * 从数据库同步房间用户到 Redis
     */
    private function syncRoomUsersFromDatabase($roomId)
    {
        echo "从数据库同步房间 {$roomId} 的用户到 Redis\n";
        
        $roomUsers = RoomUser::where('room_id', $roomId)
            ->where('status', RoomUser::STATUS_ACTIVE)
            ->get();
        
        $roomUsersKey = $this->getRoomUsersKey($roomId);
        
        foreach ($roomUsers as $roomUser) {
            $userId = $roomUser->user_id;
            $this->redis->getClient()->sadd($roomUsersKey, $userId);
            
            // 同时更新用户的房间集合
            $this->redis->getClient()->sadd($this->getUserRoomsKey($userId), $roomId);
        }
        
        $this->redis->getClient()->expire($roomUsersKey, 3600);
        echo "已同步 " . $roomUsers->count() . " 个用户到 Redis\n";
    }

    /**
     * 发送消息给连接
     */
    private function sendMessage(TcpConnection $connection, $message)
    {
        try {
            // 确保消息是数组格式
            if (!is_array($message)) {
                echo "❌ 错误: sendMessage 收到的消息不是数组格式，类型: " . gettype($message) . "\n";
                echo "消息内容: " . var_export($message, true) . "\n";
                return;
            }
            
            $jsonMessage = json_encode($message, JSON_UNESCAPED_UNICODE);
            if ($jsonMessage === false) {
                echo "❌ 消息序列化失败: " . json_last_error_msg() . "\n";
                echo "原始消息: " . var_export($message, true) . "\n";
                return;
            }
            
            $connection->send($jsonMessage);
        } catch (\Exception $e) {
            echo "❌ 发送WebSocket消息失败: " . $e->getMessage() . "\n";
            echo "异常文件: " . $e->getFile() . "\n";
            echo "异常行号: " . $e->getLine() . "\n";
            echo "消息内容: " . var_export($message, true) . "\n";
        }
    }

    /**
     * 自动订阅房间的所有活跃用户（用于消息广播时发现房间未订阅的情况）
     */
    /**
     * 自动订阅房间的所有活跃用户（当房间不在rooms数组中时使用）
     * 注意：这个方法是针对当前进程的，只能订阅到当前进程内的连接
     * 在多进程环境下，每个 worker 进程有独立的 $this->rooms 和 $this->users
     */
    private function autoSubscribeRoomUsers($roomId)
    {
        echo "========== 自动订阅房间 {$roomId} 的所有活跃用户 ==========\n";
        global $argv;
        $workerId = isset($argv[2]) ? $argv[2] : 0;
        echo "当前进程ID: " . (getmypid() ?? '未知') . "\n";
        echo "当前 Worker ID: {$workerId}\n";
        echo "处理前 - 当前进程rooms: " . json_encode(array_keys($this->rooms)) . "\n";
        echo "处理前 - 当前进程users: " . json_encode(array_keys($this->users)) . "\n";
        
        // 获取房间内所有活跃用户
        try {
            $roomUsers = RoomUser::where('room_id', $roomId)
                ->where('status', RoomUser::STATUS_ACTIVE)
                ->get();
            
            echo "从数据库查询到 " . $roomUsers->count() . " 个活跃用户\n";
            foreach ($roomUsers as $ru) {
                echo "  用户ID: {$ru->user_id}\n";
            }
        } catch (\Exception $e) {
            echo "❌ 查询房间用户失败: " . $e->getMessage() . "\n";
            return;
        }
        
        $subscribedCount = 0;
        $notConnectedCount = 0;
        foreach ($roomUsers as $roomUser) {
            $userId = $roomUser->user_id;
            
            // 检查用户是否在当前进程内已连接
            if (isset($this->users[$userId])) {
                $connection = $this->users[$userId];
                
                // 确保房间ID是字符串类型
                $roomIdStr = (string)$roomId;
                
                // 将连接添加到房间
                if (!isset($this->rooms[$roomIdStr])) {
                    $this->rooms[$roomIdStr] = [];
                    echo "创建房间 {$roomIdStr} 的连接数组\n";
                }
                
                // 检查连接是否已在房间中
                if (!in_array($connection, $this->rooms[$roomIdStr], true)) {
                    $this->rooms[$roomIdStr][] = $connection;
                    $subscribedCount++;
                    echo "✅ 自动订阅: 用户 {$userId} 已添加到房间 {$roomIdStr}（Worker {$workerId}）\n";
                } else {
                    echo "用户 {$userId} 已在房间 {$roomIdStr} 中（Worker {$workerId}）\n";
                }
                
                // 更新用户房间映射
                if (!isset($this->userRooms[$userId])) {
                    $this->userRooms[$userId] = [];
                }
                if (!in_array($roomIdStr, $this->userRooms[$userId])) {
                    $this->userRooms[$userId][] = $roomIdStr;
                }
            } else {
                $notConnectedCount++;
                echo "⚠️ 用户 {$userId} 不在当前进程（Worker {$workerId}）内连接，可能在 other worker 进程中\n";
            }
        }
        
        echo "自动订阅完成，Worker {$workerId} 成功订阅 {$subscribedCount} 个用户，{$notConnectedCount} 个用户不在当前进程\n";
        echo "处理后 - 当前进程rooms: " . json_encode(array_keys($this->rooms)) . "\n";
        echo "处理后 - 当前进程users: " . json_encode(array_keys($this->users)) . "\n";
        
        // 重要提示：因为 WebSocket 有多个 worker 进程，用户可能连接在不同的进程中
        // 所以即使当前进程的 rooms 为空，其他进程可能有连接
        if ($subscribedCount === 0 && $roomUsers->count() > 0) {
            echo "⚠️ 警告: 房间 {$roomId} 有 {$roomUsers->count()} 个活跃用户，但 Worker {$workerId} 内没有连接\n";
            echo "这可能是因为:\n";
            echo "  1. 用户连接在其他 worker 进程中（多进程环境下正常）\n";
            echo "  2. 连接已断开但数据库记录未更新\n";
            echo "  3. 连接尚未建立或已关闭\n";
        }
    }

    /**
     * 发送错误消息
     */
    private function sendError(TcpConnection $connection, $error)
    {
        $this->sendMessage($connection, [
            'type' => 'error',
            'message' => $error,
            'timestamp' => time()
        ]);
    }
}

